package com.shujia.spark.sql

import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Row, SparkSession}

object Demo4DFonRDD {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName("onrdd")
      .master("local")
      .config("spark.sql.shuffle.partitions", "1")
      .getOrCreate()

    //导入隐式转换
    import spark.implicits._
    //导入spark所有函数
    import org.apache.spark.sql.functions._

    /**
      * 可以通过SparkSession获取SparkContext
      *
      */

    val sc: SparkContext = spark.sparkContext

    val studentRDD: RDD[String] = sc.textFile("data/students.txt")

    /**
      * rdd转换成DF
      */

    val studentTuple: RDD[(String, String, String, String, String)] = studentRDD.map(stu =>{
      val str: Array[String] = stu.split(",")
      (str(0),str(1),str(2),str(3),str(4))
    })

    //转换成RDD，并指定列名
    val studentDF: DataFrame = studentTuple.toDF("id","name","age","gender","clazz")

    //studentDF.show()
    //studentDF.printSchema()


    val studentBean = studentRDD.map(stu =>{
      val str: Array[String] = stu.split(",")
      Student(str(0),str(1),str(2).toInt,str(3),str(4))
    })

    //如果rdd类型是一个自定义的类，不需要指定类型
    val studentBeanDF: DataFrame = studentBean.toDF()
    //studentBeanDF.show()
    //studentBeanDF.printSchema()

    /**
      * DF转换成RDD
      */

    val rdd: RDD[Row] = studentBeanDF.rdd

    rdd.map(row => {

      //1.通过列名和类型取字段
      val id: String = row.getAs[String]("id")
      val name: String = row.getAs[String]("name")
      val age: Int = row.getAs[Int]("age")
      val gender: String = row.getAs[String]("gender")
      val clazz: String = row.getAs[String]("clazz")

      //2、通过下标和类型取字段
      /*val id = row.getAs[String](0)
      val name = row.getAs[String](1)
      val age = row.getAs[Int](2)
      val gender = row.getAs[String](3)
      val clazz = row.getAs[String](4)*/
      (id,name,age,gender,clazz)

    })
      .foreach(println)

  }

  case class Student(id: String,name: String,age: Int,gender: String,clazz: String)

}
